package monitor.master

import akka.actor.{Actor, ActorLogging, ActorSystem, Props}
import com.typesafe.config.ConfigFactory
import monitor.master.Master.LinkedHashMap
import monitor.model.{BackRegMsgOK, MessageRegister, MessageRegisterReport}

import scala.collection.mutable

/**
  * 主节点
  */
class Master extends Actor with ActorLogging {

  var workRegMap: mutable.Map[String, MessageRegister] = new LinkedHashMap();

  override def receive: Receive = workerNodeRegister

  /**
    * 处理注解工作节点
    * @return
    */
  def workerNodeRegister: PartialFunction[Any, Unit] = {
    // 接受客户端注册数据
    case worker: MessageRegister => {
      // 打印注册信息
      if (log.isInfoEnabled) log.info("接受到worker工作节点注册信息 : {}", worker)
      // 添加数据
      workRegMap(worker.wid) = worker
      if (log.isInfoEnabled) log.info("master接受到worker工作节点注册总数 : {}", workRegMap.size)
      // 注册成功后返回worker工作节点
      sender() ! BackRegMsgOK
      workRegMap.foreach(println)
    }
    case mgr:MessageRegisterReport => {
      //更新 工作节点 报活时间
      var register = workRegMap.getOrElse(mgr.wid,null)
      if (register != null)  {
//        register.time = mgr.time
        println(s"####### ## ${register} ####################")
        workRegMap.update(
          mgr.wid,MessageRegister(mgr.wid,register.vcores,register.memory,System.currentTimeMillis()))
      }
    }
  }
}


object Master {

  type LinkedHashMap[A, B] = scala.collection.mutable.LinkedHashMap[A, B]

  val MASTER_SYSTEM = "master_system"

  val MASTER_ACTOR_SYSTEM = "master_actor_system"

  def main(args: Array[String]): Unit = {
    println("#############################  启动 master 节点 #############################")

    val Array(host, port) = args

    /**
      * 先启动人
      */
    val s: String =
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = ${host}
         |akka.remote.netty.tcp.port = ${port}
      """.stripMargin

    val config = ConfigFactory.parseString(s)

    val mastersystem = ActorSystem(MASTER_SYSTEM, config)

    mastersystem.actorOf(Props[Master], MASTER_ACTOR_SYSTEM)

  }

}
